package com.example.wan1.bus;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Scheduler;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import io.reactivex.processors.FlowableProcessor;
import io.reactivex.processors.PublishProcessor;

public final class RxBus {
    //利用 FlowableProcessor 既可以作为观察者又可以作为被观察者来实现事件总线
    private final FlowableProcessor<Object> mBus;
    //可抛出的消费者
    private final Consumer<Throwable> mOnError = new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) {
            //打印错误
            Utils.logE(throwable.toString());
        }
    };

    //toSerialized()保证线程安全
    private RxBus() {
        mBus = PublishProcessor.create().toSerialized();
    }

    //静态内部类，返回一个Holder对象
    public static RxBus getDefault() {
        return Holder.BUS;
    }

    /***
     * 发送事件
     * @param event
     */
    public void post(final Object event) {
        post(event, "", false);
    }

    /***
     * 发送事件
     * @param event
     * @param tag 标志
     */
    public void post(final Object event, final String tag) {
        post(event, tag, false);
    }

    /***
     * 发送粘性事件
     * @param event
     */
    public void postSticky(final Object event) {
        post(event, "", true);
    }


    public void postSticky(final Object event, final String tag) {
        post(event, tag, true);
    }

    /***
     * 发送事件最终都会调用到这个方法
     * @param event 事件
     * @param tag 标志
     * @param isSticky 是否粘性
     */
    private void post(final Object event,
                      final String tag,
                      final boolean isSticky) {
        //判断是否为空。事件和标志都不能为null
        //System.out.println("" == null);//结果是false
        Utils.requireNonNull(event, tag);
        //初始化TagMessage
        TagMessage msgEvent = new TagMessage(event, tag);
        //粘性事件的缓存处理
        if (isSticky) {
            CacheUtils.getInstance().addStickyEvent(event, tag);
        }
        //发送事件
        mBus.onNext(msgEvent);
    }

    /***
     * 移除粘性事件，根据事件，标志是""
     * @param event
     */
    public void removeSticky(final Object event) {
        removeSticky(event, "");
    }

    /***
     * 移除粘性事件
     * @param event
     * @param tag
     */
    public void removeSticky(final Object event,
                             final String tag) {
        //判断是否是Null
        Utils.requireNonNull(event, tag);
        CacheUtils.getInstance().removeStickyEvent(event, tag);
    }


    public <T> void subscribe(final Object subscriber,
                              final Callback<T> callback) {
        subscribe(subscriber, "", false, null, callback);
    }

    public <T> void subscribe(final Object subscriber,
                              final String tag,
                              final Callback<T> callback) {
        subscribe(subscriber, tag, false, null, callback);
    }

    public <T> void subscribe(final Object subscriber,
                              final Scheduler scheduler,
                              final Callback<T> callback) {
        subscribe(subscriber, "", false, scheduler, callback);
    }

    public <T> void subscribe(final Object subscriber,
                              final String tag,
                              final Scheduler scheduler,
                              final Callback<T> callback) {
        subscribe(subscriber, tag, false, scheduler, callback);
    }

    public <T> void subscribeSticky(final Object subscriber,
                                    final Callback<T> callback) {
        subscribe(subscriber, "", true, null, callback);
    }

    public <T> void subscribeSticky(final Object subscriber,
                                    final String tag,
                                    final Callback<T> callback) {
        subscribe(subscriber, tag, true, null, callback);
    }

    public <T> void subscribeSticky(final Object subscriber,
                                    final Scheduler scheduler,
                                    final Callback<T> callback) {
        subscribe(subscriber, "", true, scheduler, callback);
    }

    public <T> void subscribeSticky(final Object subscriber,
                                    final String tag,
                                    final Scheduler scheduler,
                                    final Callback<T> callback) {
        subscribe(subscriber, tag, true, scheduler, callback);
    }

    /***
     * 订阅事件
     * @param subscriber 订阅者
     * @param tag 标志
     * @param isSticky 粘性
     * @param scheduler 调度线程
     * @param callback 回调
     * @param <T> 回调泛型
     */
    private <T> void subscribe(final Object subscriber,
                               final String tag,
                               final boolean isSticky,
                               final Scheduler scheduler,
                               final Callback<T> callback) {
        //检查非null
        Utils.requireNonNull(subscriber, tag, callback);
        //根据回调查询泛型类
        final Class<T> typeClass = Utils.getTypeClassFromParadigm(callback);
        //创建泛型消费者，发送事件
        final Consumer<T> onNext = new Consumer<T>() {
            @Override
            public void accept(T t) {
                callback.onEvent(t);
            }
        };
        //如果是粘性
        if (isSticky) {
            final TagMessage stickyEvent = CacheUtils.getInstance().findStickyEvent(typeClass, tag);
            if (stickyEvent != null) {
                //背压 使用BackpressureStrategy.LATEST策略
                Flowable<T> stickyFlowable = Flowable.create(new FlowableOnSubscribe<T>() {
                    @Override
                    public void subscribe(FlowableEmitter<T> emitter) {
                        emitter.onNext(typeClass.cast(stickyEvent.mEvent));
                    }
                }, BackpressureStrategy.LATEST);
                //线程调度，如果不为空，加到观察的线程
                if (scheduler != null) {
                    stickyFlowable = stickyFlowable.observeOn(scheduler);
                }
                //订阅
                Disposable stickyDisposable = FlowableUtils.subscribe(stickyFlowable, onNext, mOnError);
                //加到订阅事件中，最后用来切断事件，防止内存泄漏
                CacheUtils.getInstance().addDisposable(subscriber, stickyDisposable);
            } else {
                Utils.logW("sticky event is empty.");
            }
        }
        //非粘性事件的处理方式。正常的事件处理逻辑
        Disposable disposable = FlowableUtils.subscribe(
                toFlowable(typeClass, tag, scheduler), onNext, mOnError
        );
        //加到订阅事件中，最后用来切断事件，防止内存泄漏
        CacheUtils.getInstance().addDisposable(subscriber, disposable);
    }

    /***
     * 背压处理,转换成能识别的标志信息，过滤掉相同的信息，转化成事件，最后强转成泛型事件。加上线程调度
     * @param eventType
     * @param tag
     * @param scheduler
     * @param <T>
     * @return
     */
    private <T> Flowable<T> toFlowable(final Class<T> eventType,
                                       final String tag,
                                       final Scheduler scheduler) {
        Flowable<T> flowable = mBus.ofType(TagMessage.class)
                .filter(new Predicate<TagMessage>() {
                    @Override
                    public boolean test(TagMessage tagMessage) {
                        return tagMessage.isSameType(eventType, tag);
                    }
                })
                .map(new Function<TagMessage, Object>() {
                    @Override
                    public Object apply(TagMessage tagMessage) {
                        return tagMessage.mEvent;
                    }
                })
                .cast(eventType);
        if (scheduler != null) {
            return flowable.observeOn(scheduler);
        }
        return flowable;
    }

    /***
     * 取消订阅
     * @param subscriber
     */
    public void unregister(final Object subscriber) {
        CacheUtils.getInstance().removeDisposables(subscriber);
    }

    /***
     * 静态类中
     * 静态变量
     */
    private static class Holder {
        private static final RxBus BUS = new RxBus();
    }

    /***
     * 回调
     * @param <T>
     */
    public abstract static class Callback<T> {
        public abstract void onEvent(T t);
    }

}
